package com.zhang.first.day03;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

/**
 * @title: 自定义并行source
 * @author: zhang
 * @date: 2022/1/16 12:58
 */
public class Example5 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env
                .addSource(new RichParallelSourceFunction<String>() {

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        System.out.println("并行子任务的索引是：" + getRuntimeContext().getIndexOfThisSubtask() + ",生命周期开始");
                    }

                    @Override
                    public void run(SourceContext<String> ctx) throws Exception {
                        for (int i = 1; i < 9; i++) {
                            if (i % 2 == getRuntimeContext().getIndexOfThisSubtask()) {
                                ctx.collect("并行子任务索引是：" + getRuntimeContext().getIndexOfThisSubtask() +
                                        " 发送的数据是：" + i);
                            }
                        }
                    }

                    @Override
                    public void cancel() {

                    }

                    @Override
                    public void close() throws Exception {
                        System.out.println("并行子任务的索引是：" + getRuntimeContext().getIndexOfThisSubtask() + ",生命周期结束");
                    }
                })
                .setParallelism(2)
                //数据源的并行子任务只会将一部分数据发送到print
                .rescale()
                .print()
                .setParallelism(4);

        env.execute();

    }
}
